package com.xherorlclass.proxy.handler;

import com.xherorlclass.NettyBootstrapInitializer;
import com.xherorlclass.XherorlRpcBootstrap;
import com.xherorlclass.annotation.TryTimes;
import com.xherorlclass.compress.CompressorFactory;
import com.xherorlclass.discovery.Registry;
import com.xherorlclass.enumeration.RequestType;
import com.xherorlclass.exceptions.DiscoveryException;
import com.xherorlclass.exceptions.NetworkException;
import com.xherorlclass.protection.CircuitBreaker;
import com.xherorlclass.serialize.SerializerFactory;
import com.xherorlclass.transport.message.RequestPayload;
import com.xherorlclass.transport.message.XherorlRpcRequest;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程都封装在了invoke方法中
 * 1、 发现可用服务    2、建立连接   3、发送请求   4、得到结果
 */
@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {

    // 注册中心和一个接口和一个接口
    private Registry registry;

    private Class<?> interfaceRef;

    private String group;

    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceRef, String group) {
        this.registry = registry;
        this.interfaceRef = interfaceRef;
        this.group = group;
    }

    /**
     * 所有的方法调用，本质都会走到这里
     *
     * 不是直接利用反射调用目标方法，而是通过网络通信将方法调用请求发送给远程服务，然后等待远程服务的响应结果返回。
     *
     * 流程：
     * 1、创建一个请求对象xherorlRpcRequest，封装了要调用的方法的相关信息。
     * 2、获取一个可用的通道channel，用于与远程服务进行通信。
     * 3、将请求对象写入通道，发送给远程服务。
     * 4、在XherorlRpcBootstrap.PENDING_REQUEST中注册一个CompletableFuture对象，用于保存和处理远程服务的响应结果。
     * 5、清理线程本地变量。
     * 6、等待CompletableFuture对象完成，获取远程服务的响应结果。
     * 7、返回远程服务的响应结果。
     *
     * 不需要直接调用method.invoke()方法，而是通过网络通信实现了远程方法的调用，并将调用结果返回给调用方。
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        // 从接口中获取判断是否需要重试
        TryTimes tryTimesAnnotation = method.getAnnotation(TryTimes.class);

        // 默认值0，代表不重试
        int tryTimes = 0;
        int intervalTime = 0;
        if (tryTimesAnnotation != null){
            tryTimes = tryTimesAnnotation.tryTimes();
            intervalTime = tryTimesAnnotation.intervalTime();
        }

        while (true){
            // 什么情况下需要重试，1、异常   2、响应有问题 code == 500

            /*-------------------1、封装报文-----------------------------------*/
            RequestPayload requestPayload = RequestPayload.builder()
                    .interfaceName(interfaceRef.getName()).methodName(method.getName()).parametersType(method.getParameterTypes())
                     .parametersValue(args)
                     .returnType(method.getReturnType()).build();

            // 创建一个请求
            XherorlRpcRequest xherorlRpcRequest = XherorlRpcRequest.builder()
                .requestId(XherorlRpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                .compressType(CompressorFactory.getCompressor(XherorlRpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                .requestType(RequestType.REQUEST.getId())
                .serializeType(SerializerFactory.getSerializer(XherorlRpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                    .timeStamp(System.currentTimeMillis())
                .requestPayload(requestPayload)
                .build();

            // 2、将请求存入本地线程，需要在合适的时候 remove
            XherorlRpcBootstrap.REQUEST_THREAD_LOCAL.set(xherorlRpcRequest);

            // 3、发现服务，从注册中心拉取服务列表，并通过客户端负载均衡寻找一个可用的服务
            // 传入服务的名字,返回ip+端口
            InetSocketAddress address =
                    XherorlRpcBootstrap.getInstance().getConfiguration().getLoadBalancer().selectServiceAddress(interfaceRef.getName(), group);
            if (log.isDebugEnabled()) {
                log.debug("服务调用方，发现了服务【{}】的可用主机【{}】.",
                        interfaceRef.getName(), address);
            }

            // 4、获取当前地址所对应的断路器，如果断路器是打开的则不发送请求，抛出异常
            Map<SocketAddress, CircuitBreaker> everyIpCircuitBreaker =
                    XherorlRpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker();

            CircuitBreaker circuitBreaker = everyIpCircuitBreaker.get(address);

            if (circuitBreaker == null){
                // 创建一个新的
                circuitBreaker = new CircuitBreaker(10, 0.5F);
                // 放入到缓存中
                everyIpCircuitBreaker.put(address, circuitBreaker);
            }

            try {
                // 如果断路器是打开的，并且不是心跳检测
                if (xherorlRpcRequest.getRequestType() != RequestType.HEART_BEAT.getId() && circuitBreaker.isBreak()){
                    // 定期打开
                    Timer timer = new Timer();
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            XherorlRpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker().get(address).reset();
                        }
                    }, 5000);

                    throw new RuntimeException("当前断路器已经开启，无法发送请求");
                }

                // 使用netty连接服务器，发送 调用的 服务的名字+方法名字+参数列表，得到结果
                // 定义线程池，EventLoopGroup
                // q：整个连接过程放在这里行不行，也就意味着每次调用都会产生一个新的netty连接。如何缓存我们的连接
                // 也就意味着，每次在此处建立一个新的连接是不合适的

                // 解决方案？缓存channel，尝试从缓存中获取channel，如果未获取，则创建新的连接，并进行缓存

                // 5、尝试获取一个可用通道
                Channel channel = getAvailableChannel(address);
                if (log.isDebugEnabled()) {
                    log.debug("获取了和【{}】建立的连接通道,准备发送数据.", address);
                }

                /*
                 * ------------------同步策略-------------------------
                 */
//                ChannelFuture channelFuture = channel.writeAndFlush(new Object()).await();
                // 需要学习channelFuture的简单的api get 阻塞获取结果，getNow 获取当前的结果，如果未处理完成，返回null
//                if(channelFuture.isDone()){
//                    Object object = channelFuture.getNow();
//                } else if( !channelFuture.isSuccess() ){
//                    // 需要捕获异常,可以捕获异步任务中的异常
//                    Throwable cause = channelFuture.cause();
//                    throw new RuntimeException(cause);
//                }

                /*
                 * ------------------异步策略-------------------------
                 */
                // 6、写出报文
                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                XherorlRpcBootstrap.PENDING_REQUEST.put(xherorlRpcRequest.getRequestId(), completableFuture);

                // 这里 writeAndFlush 写出一个请求，这个请求的实例就会进入 pipeline 执行出栈一系列操作
                // 第一个出栈程序一定是将 xherorlRpcRequest --> 二进制的报文
                // 以下是异步策略：
                channel.writeAndFlush(xherorlRpcRequest).addListener((ChannelFutureListener) promise->{
                // 当前的promise将来返回的结果是什么？ writeAndFlush的返回结果
                // 一旦数据被写出去，这个promise也就结束了
                // 但是我们想要的是什么？服务端给我们的返回值，所以这里处理CompletableFuture是有问题的
                // 所以这里应该将 CompletableFuture 挂起并且暴露，并且在得到服务提供方的响应的时候调用 complete 方法
//                    if (promise.isDone()){
//                        objectCompletableFuture.complete(promise.channel());

                    // 只需要处理异常就可以了
                    if (!promise.isSuccess()) {
                        completableFuture.completeExceptionally(promise.cause());
                    }
                });
                // 7、清理 threadLocal
                XherorlRpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                // 如果没有地方处理这个 completableFuture ，这里会阻塞，等待complete方法的执行完成
                // 我们需要在哪里调用 complete 方法得到结果，很明显是在 pipeline 中最终的 handler 的处理结果

                // 8、获得响应的结果
                Object result = completableFuture.get(10, TimeUnit.SECONDS);
                // 记录成功的请求
                circuitBreaker.recordRequest();
                return result;

            }catch (Exception e){
                // 次数减一，并且等待固定时间，固定时间有一定的问题，重试风暴
                tryTimes--;
                // 记录错误的次数
                circuitBreaker.recordErrorRequest();
                try {
                    Thread.sleep(intervalTime);
                }catch (InterruptedException exception){
                    log.error("在进行重试时发生异常", exception);
                }
                if (tryTimes < 0){
                    log.error("对方法【{}】进行远程调用时，重试{}次，依然不可调用",
                            method.getName(), tryTimes, e);
                    break;
                }
                log.error("在进行第{}次重试时发生异常.", 3 - tryTimes, e);
            }
        }
        throw new RuntimeException("执行远程方法" + method.getName() + "调用失败。");
    }


    /*
    * 根据地址获取一个可用的通道
    * */
    private Channel getAvailableChannel(InetSocketAddress address) {

        // 1、尝试从缓存中获取
        // 定义线程池， EventLoopGroup
        Channel channel = XherorlRpcBootstrap.CHANNEL_CACHE.get(address);

        // 2、拿不到就建立连接
        if (channel == null){
            // await方法会阻塞，会等待连接成功再返回，netty还提供了异步处理的逻辑
            // sync 和 await 都是阻塞当前线程，获取返回值（因为连接过程是异步的，发送数据的过程也是异步的）
            // 如果发生异常，sync会主动在主线程抛出异常，await不会，异常在子线程中处理需要使用future中处理


//                    channel = NettyBootstrapInitializer.getBootstrap().connect(address).await().channel();

            // 使用addListener执行的异步处理
            CompletableFuture<Channel> channelCompletableFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener((ChannelFutureListener) promise ->{
                if (promise.isDone()){
                    // 异步的，我们已经完成
                    if (log.isDebugEnabled()){
                        log.debug("以及和【{}】成功建立了连接",address);
                    }
                    channelCompletableFuture.complete(promise.channel());
                } else if (!promise.isSuccess()) {
                    channelCompletableFuture.completeExceptionally(promise.cause());
                }
            });

            // 阻塞获取channel
            try {
                channel = channelCompletableFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("获取通道时，发生异常", e);
                throw new DiscoveryException(e);
            }

            // 缓存channel
            XherorlRpcBootstrap.CHANNEL_CACHE.put(address, channel);
        }

        if (channel == null){
            log.error("获取或建立与【{}】的通道时发生了异常",address);
            throw new NetworkException("获取通道时发生异常。");
        }
        return channel;
    }
}
